package main

import (
    "github.com/op/go-logging"
    "os"
    "runtime"
    "../common"
    "../repeater"
    "os/signal"
    "syscall"
    "gopkg.in/alecthomas/kingpin.v2"
)

func init() {
    var logBackend = logging.NewLogBackend(os.Stderr, "", 0)
    logging.SetFormatter(format)
    logging.SetBackend(logBackend)
    common.SetLogger(log)
    repeater.SetLogger(log)
    runtime.GOMAXPROCS(12)
}

var format = logging.MustStringFormatter(
    `%{color}%{time:2006-01-02 15:04:05.000} %{shortfunc} %{level:.4s} %{id:03x}%{color:reset} %{message}`)
var log = logging.MustGetLogger("golia-repeater")

func main() {
    topic := kingpin.Flag("topic", "default").String()
    cfgfile := kingpin.Flag("config-file", "config file").Default("config.yml").String()
    partition := kingpin.Flag("partition", "partition").Default("0").Int32()
    kingpin.Parse()
    cfg, err := LoadConfig(*cfgfile)
    if err != nil || cfg == nil {
        log.Errorf("parse config error %v", err)
        os.Exit(2)
    }
    log.Infof("using config: %v", *cfg)
    sender, err := repeater.NewKafkaSender(cfg.Sender, *topic, *partition)
    if err != nil || cfg.Sender["type"] != "kafka-sender" {
        log.Errorf("init sender error %v", err)
        os.Exit(2)
    }
    var rpts []repeater.Repeater
    rpts = append(rpts, sender)
    for _, v := range cfg.Repeaters {
        switch v["type"] {
        case "postgre":
            r, err := repeater.NewPostgreWriter(v)
            if err != nil {
                log.Errorf("init repeater error %v", err)
                os.Exit(2)
            }
            rpts = append(rpts, r)
        case "influx":
            r, err := repeater.NewInfluxWriter(v)
            if err != nil {
                log.Errorf("init repeater error %v", err)
                os.Exit(2)
            }
            rpts = append(rpts, r)
        case "report":
            r, err := repeater.NewReportWriter(v)
            if err != nil {
                log.Errorf("init repeater error %v", err)
                os.Exit(2)
            }
            rpts = append(rpts, r)
        //case "local":
        //    r, err := repeater.NewLocalWriter(v)
        //    if err != nil {
        //        log.Errorf("init repeater error %v", err)
        //        os.Exit(2)
        //    }
        //    rpts = append(rpts, r)
        case "cache":
            r := repeater.NewCacheWriter(v)
            rpts = append(rpts, r)
        case "carbon":
            r, err := repeater.NewCarbonWriter(v)
            if err != nil {
                log.Errorf("init repeater error %v", err)
                os.Exit(2)
            }
            rpts = append(rpts, r)
        default:
            log.Noticef("not match %v", v)
            os.Exit(1)
        }
    }
    rc := repeater.NewRepeaterController(cfg.CommitInterval, cfg.CommitPath, cfg.RestAddress)
    rc.Partition = *partition
    rc.SetupLinks(rpts...)
    if err := rc.Init(); err != nil {
        log.Errorf("init controller error %v", err)
        os.Exit(2)
    }
    go rc.Run()
    term_sys := make(chan os.Signal)
    signal.Notify(term_sys, syscall.SIGTERM, syscall.SIGQUIT, syscall.SIGINT)
    select {
    case sig := <-term_sys:
        if !rc.Stopped() {
            log.Infof("receive signal %v ,exit gracefully", sig)
            rc.Stop()
        }
    case <-rc.Stopchan():
        os.Exit(2)
    }
}
